package com.jz.pay.bo.impl;

import com.jz.pay.bo.RocketMQBO;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageBatch;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Component
public class RocketMQBOImpl implements RocketMQBO {
    @Resource
    RocketMQTemplate rocketMQTemplate;

    public void send(String topic, Message<?> message) {
        rocketMQTemplate.send(topic, message);
    }

    public void send(String topic, String body) {
        rocketMQTemplate.send(topic, build(body,null));
    }
    public void send(String topic, String body, Map<String, ?> headers) {
        rocketMQTemplate.send(topic, build(body,headers));
    }

    public void sendMessageInTransaction(String topic, String body, Map<String, ?> headers) {
        rocketMQTemplate.sendMessageInTransaction(topic, build(body,headers),null);
    }

    public void sendMessageCallback(String topic, String body, Map<String, ?> headers, RocketMQLocalRequestCallback callback) {
        rocketMQTemplate.sendAndReceive(topic, build(body, headers), callback);
    }

    public void asyncSend(String topic, String body, Map<String, ?> headers, SendCallback sendCallback) {
        rocketMQTemplate.asyncSend(topic, build(body, headers), sendCallback);
    }

    public SendResult asyncSendBatch(String topic, List<String> body, Map<String, ?> headers){
        List<Message<String>> list = new ArrayList<>();
        for (String s : body) {
            list.add(build(s, headers));
        }
        return rocketMQTemplate.syncSend(topic, list,10000);
    }

    private <T> Message<T> build(T body, Map<String, ?> headers){
        return MessageBuilder.withPayload(body).copyHeaders(headers).build();
    }

}
